package cn.com.haopy.dataes.hbase.service.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

@Service
@Slf4j
public class HBaseAdminService {

    @Autowired
    private Admin hbaseAdmin;
    //private Connection connection = null;

  /*  public HBaseAdminService(Configuration conf) throws IOException {
        connection = ConnectionFactory.createConnection(conf);
        if (Objects.nonNull(connection)) {
            System.out.println("连接成功：" + connection);
            admin = connection.getAdmin();
        }

    }
*/

    private void closeTable(Table table, ResultScanner resultScanner) {
        if (Objects.nonNull(resultScanner)) {
            try {
                resultScanner.close();
            } catch (Exception e) {
                log.error(e.toString());
            }
        }
        if (Objects.nonNull(table)) {
            try {
                table.close();
            } catch (IOException e) {
                log.error(e.toString());
            }
        }
    }

    /**
     * 创建空间
     *
     * @param spaceName
     * @return
     * @throws IOException
     */
    public boolean creatSpace(String spaceName) throws IOException {
        NamespaceDescriptor nameSpace = hbaseAdmin.getNamespaceDescriptor(spaceName);
        if (Objects.isNull(nameSpace)) {
            NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(spaceName).build();
            hbaseAdmin.createNamespace(namespaceDescriptor);
            return true;
        }

        return false;
    }

    /**
     * 获取所有空间
     *
     * @return
     * @throws IOException
     */
    public List<String> getAllSpaceNames() throws IOException {
        List<String> result = new ArrayList<>();
        NamespaceDescriptor[] nameSpaces = hbaseAdmin.listNamespaceDescriptors();
        for (NamespaceDescriptor name : nameSpaces) {
            result.add(name.getName());
        }
        return result;
    }


    /**
     * 判断表是否存在
     *
     * @param tableName 表名
     * @return true/false
     */
    public boolean isExists(String nameSpace, String tableName) {
        boolean tableExists = false;
        try {
            TableName table = TableName.valueOf(nameSpace, tableName);
            tableExists = hbaseAdmin.tableExists(table);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return tableExists;
    }

    public List<String> getAllTableNames() {
        List<String> result = new ArrayList<>();
        try {
            TableName[] tableNames = hbaseAdmin.listTableNames();
            for (TableName tableName : tableNames) {
                result.add(tableName.getNameAsString());
            }
        } catch (IOException e) {
            log.error(e.toString());
        }

        return result;
    }

    /**
     * 创建表
     *
     * @param nameSpace    空间
     * @param tableName    表名
     * @param columnFamily 列族
     * @return true/false
     */
    public boolean createTable(String nameSpace, String tableName, List<String> columnFamily) {
        return createTable(nameSpace, tableName, columnFamily, null);
    }

    /**
     * 预分区创建表
     * * @param nameSpace 空间
     *
     * @param tableName    表名
     * @param columnFamily 列族
     * @param keys         分区集合
     * @return true/false
     */
    public boolean createTable(String nameSpace, String tableName, List<String> columnFamily, List<String> keys) {
        if (!isExists(nameSpace, tableName)) {
            //列族column family
            List<ColumnFamilyDescriptor> cfDesc = new ArrayList<>(columnFamily.size());
            columnFamily.forEach(cf -> {
                cfDesc.add(ColumnFamilyDescriptorBuilder.newBuilder(
                        Bytes.toBytes(cf)).build());
            });
            //表 table
            TableDescriptor tableDesc = TableDescriptorBuilder
                    .newBuilder(TableName.valueOf(nameSpace, tableName))
                    .setColumnFamilies(cfDesc).build();
            try {
                hbaseAdmin.createTable(tableDesc);
                log.debug("create table Success!");
            } catch (IOException e) {
                log.error(e.toString());
                return false;
            }

            return true;
        } else {
            System.out.println(tableName + "is exists!!!");
            return false;
        }
    }


    /**
     * 删除表
     *
     * @param tableName 表名
     */
    public void dropTable(String nameSpace, String tableName) throws IOException {
        if (isExists(nameSpace, tableName)) {
            TableName table = TableName.valueOf(nameSpace, tableName);
            hbaseAdmin.disableTable(table);
            hbaseAdmin.deleteTable(table);
        }
    }

    /**
     * 插入数据（单条）
     *
     * @param tableName    表名
     * @param rowKey       rowKey
     * @param columnFamily 列族
     * @param column       列
     * @param value        值
     * @return true/false
     */
    public boolean putData(String nameSpace, String tableName, String rowKey, String columnFamily, String column,
                           String value) {
        if (Objects.nonNull(value)) {
            return putData(nameSpace, tableName, rowKey, columnFamily, Arrays.asList(column),
                    Arrays.asList(value));
        }
        return false;

    }

    /**
     * 插入数据（批量）
     *
     * @param tableName    表名
     * @param rowKey       rowKey
     * @param columnFamily 列族
     * @param columns      列
     * @param values       值
     * @return true/false
     */
    public boolean putData(String nameSpace, String tableName, String rowKey, String columnFamily,
                           List<String> columns, List<String> values) {
        try {
            Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            for (int i = 0; i < columns.size(); i++) {
                if (Objects.nonNull(values.get(i))) {
                    put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columns.get(i)),
                            Bytes.toBytes(values.get(i)));
                }

            }
            table.put(put);
            closeTable(table, null);
            return true;
        } catch (IOException e) {
            log.error(e.toString());
            return false;
        }
    }

    /**
     * 获取数据（全表数据）
     *
     * @param tableName 表名
     * @return map
     */
    public List<Map<String, String>> getDataLimit(String nameSpace, String tableName, Integer limit) {
        List<Map<String, String>> list = new ArrayList<>();
        try {
            Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
            Scan scan = new Scan();
            if (Objects.nonNull(limit)) {
                scan.setLimit(limit);
            }
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                HashMap<String, String> map = new HashMap<>();
                //rowkey
                String row = Bytes.toString(result.getRow());
                map.put("row", row);
                for (Cell cell : result.listCells()) {
                    //列族
                    String family = Bytes.toString(cell.getFamilyArray(),
                            cell.getFamilyOffset(), cell.getFamilyLength());
                    //列
                    String qualifier = Bytes.toString(cell.getQualifierArray(),
                            cell.getQualifierOffset(), cell.getQualifierLength());
                    //值
                    String data = Bytes.toString(cell.getValueArray(),
                            cell.getValueOffset(), cell.getValueLength());
                    map.put(family + ":" + qualifier, data);
                }
                list.add(map);
            }
            closeTable(table, resultScanner);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return list;
    }


    /**
     * 获取数据（根据传入的filter）
     *
     * @param tableName 表名
     * @param filter    过滤器
     * @return map
     */
    public List<Map<String, String>> getDataFilter(String nameSpace, String tableName, Filter filter) {
        List<Map<String, String>> list = new ArrayList<>();
        try {
            Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
            Scan scan = new Scan();
            // 添加过滤器
            scan.setFilter(filter);
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                HashMap<String, String> map = new HashMap<>();
                //rowkey
                String row = Bytes.toString(result.getRow());
                map.put("row", row);
                for (Cell cell : result.listCells()) {
                    //列族
                    String family = Bytes.toString(cell.getFamilyArray(),
                            cell.getFamilyOffset(), cell.getFamilyLength());
                    //列
                    String qualifier = Bytes.toString(cell.getQualifierArray(),
                            cell.getQualifierOffset(), cell.getQualifierLength());
                    //值
                    String data = Bytes.toString(cell.getValueArray(),
                            cell.getValueOffset(), cell.getValueLength());
                    map.put(family + ":" + qualifier, data);
                }
                list.add(map);
            }
            closeTable(table, resultScanner);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return list;
    }

    /**
     * 获取数据（根据rowkey）
     *
     * @param tableName 表名
     * @param rowKey    rowKey
     * @return map
     */
    public Map<String, String> getDataRowKey(String nameSpace, String tableName, String rowKey) {
        HashMap<String, String> map = new HashMap<>();
        try {
            Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            if (result != null && !result.isEmpty()) {

                for (Cell cell : result.listCells()) {
                    //列族
                    String family = Bytes.toString(cell.getFamilyArray(),
                            cell.getFamilyOffset(), cell.getFamilyLength());
                    //列
                    String qualifier = Bytes.toString(cell.getQualifierArray(),
                            cell.getQualifierOffset(), cell.getQualifierLength());
                    //值
                    String data = Bytes.toString(cell.getValueArray(),
                            cell.getValueOffset(), cell.getValueLength());
                    map.put(family + "." + qualifier, data);

                }
            }
            closeTable(table, null);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return map;
    }

    /**
     * 获取数据（根据rowkey，列族，列）
     *
     * @param tableName       表名
     * @param rowKey          rowKey
     * @param columnFamily    列族
     * @param columnQualifier 列
     * @return map
     */
    public String getDataColumn(String nameSpace, String tableName, String rowKey, String columnFamily,
                                String columnQualifier) {
        String data = "";
        try {
            Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier));
            Result result = table.get(get);
            if (result != null && !result.isEmpty()) {
                Cell cell = result.listCells().get(0);
                data = Bytes.toString(cell.getValueArray(), cell.getValueOffset(),
                        cell.getValueLength());
            }
            closeTable(table, null);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return data;
    }

    /**
     * 删除数据（根据rowkey）
     *
     * @param tableName 表名
     * @param rowKey    rowKey
     */
    public void deleteData(String nameSpace, String tableName, String rowKey) throws IOException {
        Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        closeTable(table, null);
    }

    /**
     * 删除数据（根据rowkey，列族）
     *
     * @param tableName    表名
     * @param rowKey       rowKey
     * @param columnFamily 列族
     */
    public void deleteData(String nameSpace, String tableName, String rowKey, String columnFamily)
            throws IOException {
        Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addFamily(columnFamily.getBytes());
        table.delete(delete);
        closeTable(table, null);
    }

    /**
     * 删除数据（根据rowkey，列族）
     *
     * @param tableName    表名
     * @param rowKey       rowKey
     * @param columnFamily 列族
     * @param column       列
     */
    public void deleteData(String nameSpace, String tableName, String rowKey, String columnFamily, String column)
            throws IOException {
        Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(columnFamily.getBytes(), column.getBytes());
        table.delete(delete);
        closeTable(table, null);
    }

    /**
     * 删除数据（多行）
     *
     * @param tableName 表名
     * @param rowKeys   rowKey集合
     */
    public void deleteData(String nameSpace, String tableName, List<String> rowKeys) throws IOException {
        Table table = hbaseAdmin.getConnection().getTable(TableName.valueOf(nameSpace, tableName));
        List<Delete> deleteList = new ArrayList<>();
        for (String row : rowKeys) {
            Delete delete = new Delete(Bytes.toBytes(row));
            deleteList.add(delete);
        }
        table.delete(deleteList);
        closeTable(table, null);
    }

    /**
     * 分区【10, 20, 30】 -> ( ,10] (10,20] (20,30] (30, )
     *
     * @param keys 分区集合[10, 20, 30]
     * @return byte二维数组
     */
    private byte[][] getSplitKeys(List<String> keys) {
        byte[][] splitKeys = new byte[keys.size()][];
        TreeSet<byte[]> rows = new TreeSet<>(Bytes.BYTES_COMPARATOR);
        for (String key : keys) {
            rows.add(Bytes.toBytes(key));
        }
        int i = 0;
        for (byte[] row : rows) {
            splitKeys[i] = row;
            i++;
        }
        return splitKeys;
    }


}

